package cn.hc.dml3;

import cn.hc.conn1.HBaseConnection2;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/10/8 18:45
 */
public class HBaseDML1 {

    public static Connection connection = HBaseConnection2.connection;

    /**
     * 插入数据
     *
     * @param namespace    命名空间
     * @param tableName    表名
     * @param rowKey       主键
     * @param columnFamily 列簇
     * @param column       列名
     * @param value        值
     */
    public static void putCell(String namespace, String tableName, String rowKey, String columnFamily, String column, String value) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));

        try {
            table.put(put);
        } catch (Exception e) {
            e.printStackTrace();
        }

        table.close();
    }

    /**
     * 读取数据
     *
     * @param namespace    命名空间
     * @param tableName    表名
     * @param rowKey       主键
     * @param columnFamily 列簇
     * @param column       列名
     */
    public static void getCells(String namespace, String tableName, String rowKey, String columnFamily, String column) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        Get get = new Get(Bytes.toBytes(rowKey));
        // 直接读取，能读到一整行数据
//        table.get(get)

        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        get.readAllVersions();

        // 直接读取，能读到一整行数据
        Result result = table.get(get);

        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            String val = new String(CellUtil.cloneValue(cell));
            System.out.println(val);
        }
    }

    /**
     * 扫描数据
     *
     * @param namespace 命名空间
     * @param tableName 表名
     * @param startRow  开始的row，包含
     * @param stopRow   结束的row，不包含
     */
    public static void scanRows(String namespace, String tableName, String startRow, String stopRow) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 不填内容 直接扫描整张表
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));
//        Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));

        try {
            ResultScanner scanner = table.getScanner(scan); // Result的数组
            for (Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    System.out.print(new String(CellUtil.cloneRow(cell)));
                    System.out.print("-");
                    System.out.print(new String(CellUtil.cloneFamily(cell)));
                    System.out.print("-");
                    System.out.print(new String(CellUtil.cloneQualifier(cell)));
                    System.out.print(": ");
                    System.out.print(new String(CellUtil.cloneValue(cell)));
                    System.out.print("\t");
                }
                System.out.println();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        table.close();
    }

    /**
     * 扫描数据
     *
     * @param namespace    命名空间
     * @param tableName    表名
     * @param startRow     开始的row，包含
     * @param stopRow      结束的row，不包含
     * @param columnFamily 列簇
     * @param column       列名
     * @param value        值
     * @throws IOException
     */
    public static void filterScan(String namespace, String tableName, String startRow, String stopRow,
                                  String columnFamily, String column, String value) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        // 不填内容 直接扫描整张表
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));
//        Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));

        // 可以添加多个
        FilterList filterList = new FilterList();

        // 添加过滤
        // (1)结果只保留单列数据
//        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
//                Bytes.toBytes(columnFamily), // 列簇
//                Bytes.toBytes(column), // 列名
//                CompareOperator.EQUAL, // 比较
//                Bytes.toBytes(value) // 值
//        );
        // (2)结果保留整行数据
        // 结果同时会保留 没有当前列的数据
        SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes(columnFamily), // 列簇
                Bytes.toBytes(column), // 列名
                CompareOperator.EQUAL, // 比较
                Bytes.toBytes(value) // 值
        );
        filterList.addFilter(columnValueFilter);

        try {
            scan.setFilter(filterList);
            ResultScanner scanner = table.getScanner(scan); // Result的数组
            for (Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    System.out.print(new String(CellUtil.cloneRow(cell)));
                    System.out.print("-");
                    System.out.print(new String(CellUtil.cloneFamily(cell)));
                    System.out.print("-");
                    System.out.print(new String(CellUtil.cloneQualifier(cell)));
                    System.out.print(": ");
                    System.out.print(new String(CellUtil.cloneValue(cell)));
                    System.out.print("\t");
                }
                System.out.println();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        table.close();
    }

    /**
     * 删除一行中的一列
     *
     * @param namespace    命名空间
     * @param tableName    表名
     * @param rowKey       主键
     * @param columnFamily 列簇
     * @param column       列名
     */
    public static void deleteColumn(String namespace, String tableName, String rowKey, String columnFamily, String column) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        // 删除一个版本
        delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        // 删除所有版本
        delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(column));

        try {
            table.delete(delete);
        } catch (Exception e) {
            e.printStackTrace();
        }

        table.close();
    }

    public static void main(String[] args) throws IOException {
        // 测试添加数据
        putCell("bigdata", "student", "1001", "info", "name", "zsf");
        putCell("bigdata", "student", "1006", "info", "age", "14");
        putCell("bigdata", "student", "2001", "info", "name", "zs");
        putCell("bigdata", "student", "2001", "info", "name", "ls");
        putCell("bigdata", "student", "2001", "info", "name", "ww");

        // 测试读取数据
        getCells("bigdata", "student", "2001", "info", "name");

        // 测试扫描数据
        scanRows("bigdata", "student", "1001", "2002");

        filterScan("bigdata", "student", "1001", "2002", "info", "name", "zsf");

        // 测试删除
        System.out.println("测试删除-----------");
        deleteColumn("bigdata", "student", "2001", "info", "name");
        scanRows("bigdata", "student", "1001", "2002");

        System.out.println("其他代码");

        HBaseConnection2.closeConnection();
    }
}
